package com.yl.flink.processor;

import cn.hutool.core.util.StrUtil;
import com.yl.constant.Const;
import com.yl.entity.cdc.BaseTargetType;
import com.yl.entity.cdc.MysqlCdcEntity;
import com.yl.util.ClassUtil;
import com.yl.util.FlinkUtil;
import com.yl.util.InfluxUtil;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.influxdb.InfluxDBConfig;
import org.apache.flink.util.Collector;
import org.influxdb.InfluxDB;
import org.influxdb.dto.Query;

import java.util.Optional;

public class InitCQFunc extends ProcessFunction<MysqlCdcEntity, Object> {

    private transient InfluxDB influxDBClient;
    private final InfluxDBConfig influxDBConfig;

    public InitCQFunc(ParameterTool params) {
        this.influxDBConfig = FlinkUtil.getInfluxDBConfig(params);
    }

    /**
     * 初始化资源配置
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        this.influxDBClient = InfluxUtil.connect(this.influxDBConfig);
    }

    /**
     * 创建各类型统计每小时数据条数的连续查询
     */
    @Override
    public void processElement(MysqlCdcEntity value, ProcessFunction<MysqlCdcEntity, Object>.Context ctx, Collector<Object> out) throws Exception {
        Optional.ofNullable(ClassUtil.cast(value, BaseTargetType.class))
                .ifPresent(baseTargetType -> {
                    String db = this.influxDBConfig.getDatabase();
                    String typeTag = baseTargetType.getTag().toLowerCase();
                    String sql = StrUtil.format(Const.INFLUXDB_SQL_CQ, typeTag, db, typeTag, typeTag);
                    this.influxDBClient.query(new Query(sql, db));
                });
    }

    @Override
    public void close() {
        if (this.influxDBClient.isBatchEnabled()) {
            this.influxDBClient.disableBatch();
        }
        this.influxDBClient.close();
    }
}
